package me.haoliang;

import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.Arrays;
import java.util.List;

/**
 * Created by pulangke on 2017/5/23.
 * <p>
 * see evernote Reactor by Example
 *
 * @
 */
public class FluxDemo {

    private static Logger logger = LoggerFactory.getLogger(FluxDemo.class);

    private static List<String> words = Arrays.asList(
            "the",
            "quick",
            "brown",
            "fox",
            "jumped",
            "over",
            "the",
            "lazy",
            "dog"
    );


    @Test
    public void fluxSubscribe() {
        Flux<String> manyWord = Flux.fromIterable(words);
        manyWord.subscribe(x -> logger.info("{}", x));
    }


    @Test
    public void findingMissingLetter() {
        Flux<String> manyLetterf = Flux.fromIterable(words)
                .flatMap(word -> Flux.fromArray(word.split("")))
                .distinct()
                .sort()
                .zipWith(Flux.range(1, Integer.MAX_VALUE), (word, count) -> String.format("%2d.%s", count, word));
        manyLetterf.subscribe(x -> logger.info("{}", x));
    }


    @Test
    public void restoringMissingLetter() {
        Mono<String> missingWord = Mono.just("s");
        Flux<String> flux = Flux.fromIterable(words)
                .flatMap(word -> Flux.fromArray(word.split("")))
                .concatWith(missingWord)
                .distinct()
                .sort()
                .zipWith(Flux.range(1, Integer.MAX_VALUE), (word, count) -> String.format("%2d.%s", count, word));
        flux.subscribe(x -> logger.info("{}", x));
    }

    @Test
    public void shortCircuit() {
        Flux<String> helloPauseWorld = Mono.just("hello")
                .concatWith(Mono.just("world").delaySubscription(Duration.ofMillis(500)));

        helloPauseWorld.subscribe(x -> logger.info("{}", x));

    }

    @Test
    public void blocks() {
        Flux<String> helloPauseWorld = Mono.just("Hello")
                .concatWith(Mono.just("world").delaySubscription(Duration.ofMillis(500)));

        helloPauseWorld.toStream()
                .forEach(System.out::println);
    }

    @Test
    public void firstEmitting() {
        Mono<String> a = Mono.just("oops I'm late")
                .delaySubscription(Duration.ofMillis(400));
        Flux<String> b = Flux.just("let's get", "the party", "started")
                .delaySubscription(Duration.ofMillis(400));

        Flux.firstEmitting(a, b)
                .toIterable()
                .forEach(System.out::println);
    }


    @Test
    public void testHello() {

        Mono<String> body = Mono.just("luhui");
        Flux<String> flux = Mono.just("Hey mister ")
                .concatWith(
                        body.flatMapMany(sir -> Flux.fromArray(sir.split("")))
                                .map(String::toUpperCase)
                                .take(1)
                ).concatWith(Mono.just(". how are you?"));
        flux.subscribe(x -> logger.info("{}", x));


        Mono<String> mono = flux.reduce((x, y) -> x = x.concat(y));
        mono.subscribe(x -> logger.info("{}", x));

    }
}
